package com.business.etl.flow.process;

import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Multimap;
import com.component.api.model.ProcessResult;

import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.LockSupport;

/**
 * 描述：
 * 过程处理对象
 * 管理中间数据和运行中的线程依赖
 *  运行结果数据缓存 根据节点绑定
 *  这里封装一个对象 方便后面扩展
 *  大数据量应该采用分布式框架处理 这里临时解决 只能处理小数据
 * @author xianggj
 * @Date 2021/10/15 17:56
 **/
public class ProcessHandler {

    public ProcessHandler(String flowId) {
        this.flowId = flowId;
    }

    private String flowId;

    public String getFlowId() {
        return flowId;
    }

    /**
     * 节点数据缓存
     * key： nodeId
     */
    private  Map<String, ProcessResult> nodeCache = new ConcurrentHashMap<>();

    /**
     * 单节点允许的最大执行时长，超过自动释放线程
     */
    public  final long MAX_TIME = 12*60*60*1000L;

    /**
     * 记录当前线程在等待那个节点
     * key：线程依赖的节点
     * Value: 线程对象
     */
    private Multimap<String, Thread> relyOnThreadMap = ArrayListMultimap.create();

    /**
     * 线程缓存记录缓存 只记录正在运行中的线程
     * key： nodeId
     */
    private  Map<String, Thread> threadMap = new ConcurrentHashMap<>();

    public  ProcessResult getResult(String nodeId) {
        return nodeCache.get(nodeId);
    }

    public synchronized void putValue(String nodeId, ProcessResult result) {
        nodeCache.put(nodeId, result);
        if (relyOnThreadMap.containsKey(nodeId)){
            Collection<Thread> threads = relyOnThreadMap.get(nodeId);
            for (Thread thread : threads) {
                //线程已经被中断
                if (thread.isInterrupted()){
                    continue;
                }
                System.out.println("线程[" +thread.getName()+":"+thread.getId()+"]恢复运行.");
                LockSupport.unpark(thread);
            }
        }
    }

    /**
     * 这里存在一个线程问题，当nodeCache添加的时候 relyOnThreadMap也在添加
     * @param nodeId
     * @param thread
     * @return
     */
    public synchronized boolean putRelyNoThread(String nodeId, Thread thread) {
        if (!nodeCache.containsKey(nodeId)){
            relyOnThreadMap.put(nodeId, thread);
            return true;
        } else {
            return false;
        }
    }

    public void putThread(String nodeId, Thread thread) {
        threadMap.put(nodeId, thread);
    }

    public boolean isCompleted(String nodeId) {
        return nodeCache.containsKey(nodeId);
    }

    public boolean isRunning(String nodeId) {
        return threadMap.containsKey(nodeId);
    }

    /**
     * 释放数据
     * @param nodeId
     */
    public void clear(String nodeId){
        System.out.println(String.format("释放节点【%s】的数据", nodeId));
        nodeCache.remove(nodeId);
    }

    /**
     * 释放线程 除了主动停止 其他时候不能释放
     * @param nodeId
     */
    public void release(String nodeId){
        System.out.println(String.format("释放节点【%s】相关的线程" , nodeId));
        if (threadMap.containsKey(nodeId)){
            Thread thread = threadMap.get(nodeId);
            if (!thread.isInterrupted()){
                thread.interrupt();
            }
            threadMap.remove(nodeId);
        }
    }

    /**
     * 使线程中断
     * @param nodeId
     */
    public void stop(String nodeId){
        System.out.println(String.format("释放节点【%s】的数据", nodeId));
        nodeCache.remove(nodeId);
        if (relyOnThreadMap.containsKey(nodeId)){
            relyOnThreadMap.removeAll(nodeId);
        }
        release(nodeId);
    }

}
